# -*- coding: utf-8 -*-
# @Time    : 2025/8/5 14:37
# @Author  : Zhaoxu
# @File    : t2v.py
# @Software: PyCharm
# -*- coding: utf-8 -*-
"""
GPT → 讯飞 TTS → PyAudio 实时播放
-------------------------------------------------
依赖：
    pip install openai websocket-client pyaudio
    # Windows 若安装 pyaudio 困难，可先
    #   pip install pipwin
    #   pipwin install pyaudio
说明：
    1. GPT 流式输出 → 本地切块（≥50 字符或句号等）→ 放入队列
    2. 消费线程对每块文字调用讯飞 TTS (websocket)
    3. 收到每帧音频立即写入 PyAudio Stream，边合成边播放
-------------------------------------------------
"""
import queue
import threading
import time
import os
import base64
import json
import hashlib
import hmac
from urllib.parse import urlencode
from datetime import datetime
from wsgiref.handlers import format_date_time
import websocket
import pyaudio
from openai import OpenAI


# =============== GPT → 队列 =============== #
def gpt_stream_to_queue(q: queue.Queue):
    """
    GPT 流式输出 → 本地分段 → 入队
    """
    client = OpenAI(
        api_key="sk-9bc1ce71791b49f083e625f2f2788b2d",     # ★换成自己的 DeepSeek/ChatKey
        base_url="https://api.deepseek.com"
    )
    resp = client.chat.completions.create(
        model="deepseek-chat",
        messages=[
            {"role": "system", "content": "You are a helpful assistant, response in 30 words and no emoji"},
            {"role": "user", "content": "hello ,may i ask you a question"}
        ],
        stream=True
    )

    buffer = ""
    for part in resp:
        delta = part.choices[0].delta
        if not getattr(delta, "content", None):
            continue
        token = delta.content
        print(token, end="", flush=True)          # 终端同步看到文本
        buffer += token
        if len(buffer) >= 20 or buffer.endswith(('.', '!', '?', '。', '！', '？')):
            q.put(buffer)
            buffer = ""
    if buffer.strip():
        q.put(buffer)
    q.put(None)     # 结束信号


# =============== 讯飞 TTS websocket 工具 =============== #
class Ws_Param:
    def __init__(self, appid, apikey, apisecret, text):
        self.APPID = appid
        self.APIKey = apikey
        self.APISecret = apisecret
        self.Text = text

        self.CommonArgs = {"app_id": self.APPID}
        # raw = 16bit,16000Hz,1ch ；可改 vcn 声音、人声类型
        self.BusinessArgs = {"aue": "raw",
                             "auf": "audio/L16;rate=16000",
                             "vcn": "x4_yezi",
                             "tte": "utf8"}
        self.Data = {"status": 2,
                     "text": str(base64.b64encode(self.Text.encode('utf-8')), "utf8")}

    def create_url(self):
        url = "wss://tts-api.xfyun.cn/v2/tts"
        now = datetime.now()
        date = format_date_time(time.mktime(now.timetuple()))
        signature_origin = f"host: ws-api.xfyun.cn\n" \
                           f"date: {date}\n" \
                           f"GET /v2/tts HTTP/1.1"
        sign = hmac.new(self.APISecret.encode(), signature_origin.encode(),
                        digestmod=hashlib.sha256).digest()
        sign = base64.b64encode(sign).decode()
        auth_origin = f'api_key="{self.APIKey}", algorithm="hmac-sha256", ' \
                      f'headers="host date request-line", signature="{sign}"'
        authorization = base64.b64encode(auth_origin.encode()).decode()
        return url + "?" + urlencode({"authorization": authorization,
                                      "date": date,
                                      "host": "ws-api.xfyun.cn"})


def tts_stream_and_play(appid, apikey, apisecret, text):
    """
    对 text 进行 TTS；边接收边播放（PyAudio），阻塞直到该段播放完毕
    """
    p = pyaudio.PyAudio()
    stream = p.open(format=pyaudio.paInt16,
                    channels=1,
                    rate=16000,
                    output=True)

    done_flag = threading.Event()

    def on_message(ws, message):
        msg = json.loads(message)
        audio_bytes = base64.b64decode(msg["data"]["audio"])
        stream.write(audio_bytes)                  # ★ 直接播
        if msg["data"]["status"] == 2:
            ws.close()                             # 主动关闭，触发 on_close

    def on_error(ws, error):
        print("ws error:", error)

    def on_close(ws, close_status_code, close_msg):
        # 关闭音频流
        stream.stop_stream()
        stream.close()
        p.terminate()
        done_flag.set()

    def on_open(ws):
        def run(*args):
            req = {"common": wsParam.CommonArgs,
                   "business": wsParam.BusinessArgs,
                   "data": wsParam.Data}
            ws.send(json.dumps(req))
        threading.Thread(target=run).start()

    wsParam = Ws_Param(appid, apikey, apisecret, text)
    ws = websocket.WebSocketApp(wsParam.create_url(),
                                on_open=on_open,
                                on_message=on_message,
                                on_error=on_error,
                                on_close=on_close)
    websocket.enableTrace(False)
    ws.run_forever(sslopt={"cert_reqs": 0})
    done_flag.wait()           # 等播放完毕


# =============== 消费队列线程 =============== #
def consumer_tts_play(q: queue.Queue, appid, apikey, apisecret):
    idx = 1
    while True:
        txt = q.get()
        if txt is None:
            break
        print(f"\n\n>>> 第 {idx} 块文本（{len(txt)} 字）开始合成并立即播放 …")
        tts_stream_and_play(appid, apikey, apisecret, txt)
        idx += 1
    print("\n===== 全文播放结束 =====")


# =============== main =============== #
def main():
    # ★换成自己的讯飞账号信息
    # 讯飞配置（请替换成自己账号）
    APPID = "9646138c"
    APIKEY = "ea89fb89877aded124af01b32cb14324"
    APISECRET = "MzlhYmQyMjI5ODYzYzU2ODM2YWMxYWNk"
    q = queue.Queue()

    # 线程A ：GPT 生产文本块
    threading.Thread(target=gpt_stream_to_queue,
                     args=(q,), daemon=True).start()

    # 主线程：消费→TTS→播放
    consumer_tts_play(q, APPID, APIKEY, APISECRET)


if __name__ == "__main__":
    main()